package cn.creditease.std.consumer;


import cn.creditease.std.context.JobAlloterContext;
import cn.creditease.std.dao.JobDAO;
import cn.creditease.std.enums.JobStatusEnums;
import cn.creditease.std.enums.TryConsumeStatusEnums;
import cn.creditease.std.model.Job;
import cn.creditease.std.monitor.AbstractRepairMonitor;
import cn.creditease.std.monitor.impl.RepairCheckJobMonitor;
import cn.creditease.std.monitor.impl.RepairExpirePrecessingJobMonitor;
import cn.creditease.std.utils.ThreadFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

/**
 * @Description: 默认实现的job抽象消费处理类
 * @Author: kai.yang
 * @Date: 2019-08-05 16:27
 */
public abstract class AbstactJobConsumeWorker<T> implements JobConsumer<T> {

    JobDAO jobDao = JobDAO.newInstance();


    /**
     * 默认没过时间范围内，会一直补偿
     */
    private static final int DEFAULT_MAX_RETRY_COUNT = Integer.MAX_VALUE;


    private static final Logger logger = LoggerFactory.getLogger(AbstactJobConsumeWorker.class);

    private List<AbstractRepairMonitor> monitors = new ArrayList<>();

    public AbstactJobConsumeWorker() {
        //注册对应的子类
        JobConsumerRegistory.registry(getJobName(), this);

        init();


    }

    /**
     * 当前服务器当前任务消费者数量
     */
    private AtomicInteger consumerCount = null;

    /**
     * 初始化基本配置，幂等操作，每个任务组第一次生效
     */
    private void init() {
        if (consumerCount == null) {
            synchronized (this) {
                if (consumerCount == null) {
                    try {
                        //初始化资源池大小
                        consumerCount = new AtomicInteger(
                            JobAlloterContext.getContext().getJobConsumerConfigMap().get(getJobName())
                                .getMaxConsumerCountByJobGroupPerServer());

                        //初始化补偿检测器

                        monitors.add(new RepairCheckJobMonitor(getJobName()));
                        monitors.add(new RepairExpirePrecessingJobMonitor(getJobName()));

                        if (!CollectionUtils.isEmpty(monitors)) {
                            monitors.stream().forEach(monitor -> monitor.check());
                        }
                    } catch (Exception e) {
                        logger.info("分发管理器还未初始化稍后重试");
                    }

                }
            }
        }
    }

    private boolean isLeisure() {
        //只有服务器恢复到最大容量时才去取下一批任务数据
        return consumerCount.get() == JobAlloterContext.getContext().getJobConsumerConfigMap().get(getJobName())
            .getMaxConsumerCountByJobGroupPerServer();
    }

    @Override
    public void consume(String jobName) {
        consume(null, jobName);
    }


    @Override
    public void consume(String jobId, String jobName) {
        retryPopJobs(jobId, jobName, 1, null,false);
    }

    @Override
    public void processJob(Job job, Function calBack) {
        boolean result = false;
        try {
            if (logger.isDebugEnabled()) {
                logger.debug("消费队列job={}", job);
            }
            //1. 根据jobId获取对应任务数据,判断任务状态
            //（1）如果任务已经完成，直接返回
            //（2）如果任务为中间状态，判断是否超过处理时长

            if (!job.getJobStatus().equals(JobStatusEnums.PROCESS)) {
                logger.info("不合法的任务状态={},id={},请限制try方法", job.getJobStatus(), job.getJobId());
                return;
            }

            logger.info("执行任务jobId的={}", job.getJobId());

            //2. 调用业务实际查询业务数据方法
            List<T> reader;
            if (job == null) {
                return;
            } else {
                reader = doReader(job.getPageNo(), job.getPageSize());
            }

            //3. 调用处理数据方法
            if (CollectionUtils.isEmpty(reader)) {
                return;
            } else {
                result = doProcess(reader);
            }


        } catch (Throwable e) {
            logger.info("消费任务数据异常", e);
        } finally {
            try {
                //4.根据操作结果修改任务状态,并释放资源
                afterPrecessJob(job, result, calBack);
            } catch (Exception e) {
                //do nothing
            }

        }


    }

    /**
     * 执行任务对前置处理
     * 1.如果为INIT状态
     * 将状态改为PROCESS状态，成功返回true
     * 2.如果为PROCESS状态
     * 2.1 判断是否超过超时时间，如果超时，修改为INIT，并retryCount加1
     */
    @Override
    public TryConsumeStatusEnums tryProcessJob(Job job, boolean isSuccessCallback) {

        init();

        boolean needRelease = false;
        try {
            if (job == null) {
                return TryConsumeStatusEnums.EXCEPTION;
            } else {
                //1.
                int i = consumerCount.get();
                if (i > 0) {
                    boolean tryConsume = consumerCount.compareAndSet(i, i - 1);
                    if (!tryConsume) {
                        //获取失败后，此处自旋重试,直到获取成功，或consumerCount为0,
                        tryProcessJob(job, isSuccessCallback);
                    }
                } else {
                    return TryConsumeStatusEnums.POOL_FULL;
                }

                if (job.getJobStatus().equals(JobStatusEnums.INIT)) {

                    //如果修改成功，返回获取锁成功
                    if (jobDao
                        .updateStatusAndConsumerId(job.getId(),
                            JobAlloterContext.getContext().getCustomerId(),
                            JobStatusEnums.INIT.toString(), JobStatusEnums.PROCESS.toString())
                        > 0) {
                        job.setJobStatus(JobStatusEnums.PROCESS);
                        return TryConsumeStatusEnums.TRY_LOCK_SUCCESS;
                    } else {
                        if (logger.isDebugEnabled()) {
                            logger.debug("任务处理乐观锁消任务jobId={}", job.getId());
                        }
                        //获取服务资源成功，但数据库乐观锁失败，需要释放资源
                        needRelease = true;
                        return TryConsumeStatusEnums.TRY_LOCK_FAIL;
                    }
                } else {
                    if (logger.isDebugEnabled()) {
                        logger.debug("此处不应存在非INIT状态任务id={},status={}", job.getJobId(), job.getJobStatus());
                    }
                }
            }
            return TryConsumeStatusEnums.EXCEPTION;
        } finally {
            if (needRelease) {
                release();
            }
        }
    }


    /**
     * 如果任务执行成功，将任务状态状态修改
     */
    @Override
    public void afterPrecessJob(Job job, boolean result, Function<String, Boolean> calBack) {
        try {
            if (job != null) {
                //1.任务成功，修改状态为成功
                if (result) {
                    jobDao
                        .updateStatus(job.getId(), JobStatusEnums.PROCESS.toString(),
                            JobStatusEnums.SUCCESS.toString());
                } else {
                    if (job.getRetryCount() < DEFAULT_MAX_RETRY_COUNT) {
                        //2。任务失败，且未超次数限制，置为init等下次执行
                        jobDao.updateStatusAndAddCount(job.getId(), JobStatusEnums.PROCESS.toString(),
                            JobStatusEnums.INIT.toString());
                    } else {

                        //2。任务失败，且超次数限制
                        jobDao.updateStatusAndAddCount(job.getId(), JobStatusEnums.PROCESS.toString(),
                            JobStatusEnums.FAIL.toString());
                    }
                }
            }
        } finally {
            release();

            // 任务执行完了，回调任务
            calBack.apply(job.getJobGroupName());


        }
    }

    @Override
    public boolean release() {
        consumerCount.getAndIncrement();
        return true;
    }


    /**
     *    不断自旋，读取可用job
     * @param jobId 一次job的统一id
     * @param jobName 一组job的统一名称
     * @param retryCount 已重试获取任务次数，有失败最大限制，默认最大失败5次
     * @param limitStartId id索引限制，用于获取任务时，乐观锁失败，获取下一组任务将会直接从下一页开始获取，避免当前页太多冲突
     * @param succesCallback
     */
    private void retryPopJobs(String jobId, String jobName, Integer retryCount,Long limitStartId, boolean succesCallback) {

        if (logger.isDebugEnabled()) {
            logger.debug("消费队列job={},jobName={}", jobId, jobName);
        }

        if (retryCount > 5) {
            logger.warn("重试次数太多了={}，居然还没获取到锁数据，不取了，如果还有只能补偿了", retryCount);
            return;
        }
        /**
         * 每次从数据库最多拿任务池最大容量任务，避免过多获取无法消费任务,每次会根据retryCount进行跳页查询
         */
        int limit = JobAlloterContext.getContext().getJobConsumerConfigMap().get(getJobName())
            .getMaxConsumerCountByJobGroupPerServer();

        //此处兼容根据id进行实时分发模式，和根据jobName补偿模式,
        //再后续重试中通过limitStartId现在id索引，进行分页滑动查询，减少乐观锁冲突
        List<Job> jobs;
        if (!StringUtils.isEmpty(jobId)) {
            //根据jobId获取2天内，待处理到一组任务
            jobs = jobDao.queryByJobIdAndStatusAndCreateTimeLimit(jobId,
                limitStartId, limit,
                JobStatusEnums.INIT);
        } else {
            jobs = jobDao.queryByJobNameAndStatusAndCreateTimeLimit(jobName,
                limitStartId, limit,
                JobStatusEnums.INIT);
        }

        //如果待处理任务不为空，进行获取任务执行锁尝试
        if (!CollectionUtils.isEmpty(jobs)) {
            for (Job job : jobs) {
                TryConsumeStatusEnums tryConsumeStatusEnums = this.tryProcessJob(job, false);
                //获取锁成功后执行本组数据
                switch (tryConsumeStatusEnums) {
                    case TRY_LOCK_SUCCESS:
                        ThreadFactory.startNewThread(() -> {
                            //执行消费者逻辑
                            this.processJob(job, jobNameStr -> {
                                if (jobNameStr != null && isLeisure()) {

                                    //回调进行处理下一组数据，避免任务大部分空闲
                                    retryPopJobs(jobId, jobName, 1, jobs.get(jobs.size()-1).getId(),true);
                                }
                                return true;
                            });
                        });
                        break;
                    case TRY_LOCK_FAIL:
                        //获取锁失败问题，将会重试下一组
                        if (succesCallback &&  jobs.size() < limit){
                            if(logger.isDebugEnabled()){
                                logger.debug("已经是最后一组，不用再试了");
                            }
                        }else {
                            int nextCount = ++retryCount;
                            ThreadFactory.startNewThread(() ->
                                retryPopJobs(jobId, jobName, nextCount, jobs.get(jobs.size()-1).getId(),false)
                            );
                        }
                        break;
                    case POOL_FULL:
                    case EXCEPTION:
                        if (logger.isDebugEnabled()) {
                            logger.debug("消费者已满,消费停止");
                        }
                    default:
                        return;
                }


            }

        } else {
            if (logger.isDebugEnabled()) {
                logger.debug("消费队列job={},jobName={}任务数获取为0", jobId, jobName);
            }
        }
    }


}
